package cn.ctodb.im.server.kafka;

import cn.ctodb.im.dto.AbstractDTO;
import cn.ctodb.im.kafka.KafkaUtils;
import cn.ctodb.im.server.conf.ApplicationProperties;
import cn.ctodb.im.server.peer.PeerManager;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Collections;

@Component
public class KafkaConsumer implements DisposableBean {

    private Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);

    @Autowired
    private PeerManager peerManager;
    @Autowired
    private ApplicationProperties applicationProperties;
    @Autowired
    private KafkaAdmin kafkaAdmin;

    @KafkaListener(topics = "${application.kafka-topic}")
    public void text(ConsumerRecord<?, String> record) {
        try {
            String str = record.value();
            logger.debug("mq : {}", str);
            AbstractDTO textReq = KafkaUtils.getDate(record);
            peerManager.get(KafkaUtils.getChannelId(record)).getChannel().writeAndFlush(textReq.toProto());
        } catch (IOException e) {
            logger.error(e.getMessage(), e);
        } catch (ClassNotFoundException e) {
            logger.error(e.getMessage(), e);
        }
    }

    @Override
    public void destroy() {
        logger.debug("kafka destroy : 貌似不好使啊......");
        AdminClient.create(kafkaAdmin.getConfig()).deleteTopics(Collections.singleton(applicationProperties.getKafkaTopic()));
    }
}